package com.fanxl.flink.datastream.windows;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.types.LongValue;
import org.apache.flink.types.NullValue;

/**
 * @author Dell
 * @Title:
 * @Description: 请写注释类
 * @date 2022/4/11
 */
public class NotOccur {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        test01(env);
        env.execute("WindowApp");


    }

    public static void test01(StreamExecutionEnvironment env) {
        SingleOutputStreamOperator<Integer> filter = env.socketTextStream("10.215.15.200", 9528)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.parseInt(value);
                    }
                }).filter(new FilterFunction<Integer>() {
                    @Override
                    public boolean filter(Integer value) throws Exception {
                        return value > 10;
                    }
                });

        IterativeStream<Integer> iterate = filter.iterate(5000);
        if(iterate == null ){
            System.out.println("不发生时间");
        }else{
            iterate.print("daying");
        }
        filter
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum(0)
                .print();

    }

}
